package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MessageQueueException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import org.jetbrains.annotations.NotNull;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/*
*  通过这个类，对硬盘上的消息进行处理
*/
public class MessageFileManager {

    // 因这个类只在这个文件中使用到，所以为了便捷与简洁 定义成内部类使用
    static public class Stat{
        public int totalCount; // 总消息数量
        public int validCount; // 有效消息数量
    }

    public void init() {
        // 暂时不需要进行初始化操作，以备后续扩展
    }

    // 获取指定消息文件所在的目录和文件名
    private String getQueueDir(String queueName) { // 用来获取指定消息队列的 消息文件所在队列路径
        return "./data/" + queueName;
    }

    private String getQueueDataPath(String queueName){ // 用来获取该队列的 消息数据文件的路径
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    private String getQueueStatPath(String queueName){// 用来获取该队列中的 消息统计文件路径
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName){
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner = new Scanner(inputStream); // 因当前的消息统计文件是文本文件，所以使用 Scanner 来读取文件内容
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        }  catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat){
        //  OutputStream 打开文件, 默认情况下, 会直接把原文件清空. 此时相当于新的数据覆盖了旧的.
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            // 使用 PrintWriter 写文件
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush(); // 确保消息及时刷新到内存中去
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建队列对应的文件和目录
    public void createQueueFile(String queueName) throws IOException {
        // 创建队列的消息目录
        File queueDir = new File(getQueueDir(queueName));
        if (!queueDir.exists()){
            boolean ret = queueDir.mkdirs();
            if (!ret){
                throw new IOException("创建队列目录失败! queueDir = " + queueDir.getAbsolutePath());
            }
        }

        // 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()){
            boolean ret = queueDataFile.createNewFile();
            if (!ret){
                throw new IOException("创建队列数据文件失败! queueDataFile = " + queueDataFile.getAbsolutePath());
            }
        }

        // 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()){
            boolean ret = queueStatFile.createNewFile();
            if (!ret){
                throw new IOException("创建队列统计数据文件失败! queueStatFile = " + queueStatFile.getAbsolutePath());
            }
        }

        // 给消息统计文件，设定初始值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    // 检查队列的目录和文件是否存在
    public boolean checkQueueAndFileExits(String queueName) {
//        // 检查队列目录是否存在
//        File queueDir = new File(getQueueDir(queueName));
//        if (!queueDir.exists()){
//           return false;
//        }
        // 检查队列数据文件是否存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()){
            return false;
        }
        // 检查队列数据统计是否存在
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    // 删除队列 的目录和文件
    public void deleteQueueFile(String queueName) throws IOException {
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean retData = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean retStat = queueStatFile.delete();
        File queueDir = new File(getQueueDir(queueName));
        boolean retDir = queueDir.delete();
        if (!retData || !retStat || !retDir){
            throw new IOException("删除队列目录和文件失败! queueDir = "+ queueDir.getAbsolutePath());
        }
    }

    // 这个方法用来把一个新 消息（Message）放到对应的队列中去
    public void sendMessage(@NotNull MSGQueue queue, Message message) throws MessageQueueException, IOException {
            // 检查当前要写入的队列对应的文件是否存在
            if (!(checkQueueAndFileExits(queue.getName()))){
                throw new MessageQueueException("[MessageFileManager] 队列对应的文件不存在! queueName = " + queue.getName());
            }

            // 把 Message 对象，进行序列化，转成二进制的字节数组
            byte[] messageBinary = BinaryTool.toBytes(message);
            synchronized (queue) { // 通过对 同一个队列加锁， 保证在读写文件多线程操作时，数据的一致性和准确性
                // 获取到当前的队列数据文件的长度，计算出该 Message 对象的 offsetBeg 和 offsetEnd
                // 通过方法 queueDataFile.length() 就能获取到文件的长度. 单位字节.
                File queueDataFile = new File(getQueueDataPath(queue.getName())); // 获取到 队列的数据文件
                message.setOffsetBeg(queueDataFile.length() + 4); // 把新的 Message 数据, 写入到队列数据文件的末尾. 此时 Message 对象的 offsetBeg , 就是当前文件长度 + 4
                message.setOffsetEnd(queueDataFile.length() + messageBinary.length + 4); // offsetEnd 就是当前文件长度 + 4 + message 自身长度.

                // 写入消息到数据文件中，是追加到数据文件末尾
                try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                    try (DataOutputStream dataOutputStream =new DataOutputStream(outputStream)){
                        // 接下来要先写当前消息的长度, 占据 4 个字节的~~
                        dataOutputStream.writeInt(messageBinary.length);
                        // 写入消息本体
                        dataOutputStream.write(messageBinary);

                    }
                }

                // 更新消息统计文件
                Stat stat = readStat(queue.getName());
                stat.totalCount += 1;
                stat.validCount += 1;
                writeStat(queue.getName(), stat);
            }


    }

    // 删除队列数据文件中的消息 / 这里的删除是逻辑删除, 也就是把硬盘上存储的这个数据里面的那个 isValid 属性, 设置成 0
    // 此处这个参数中的 message 对象, 必须得包含有效的 offsetBeg 和 offsetEnd
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) { // 加锁 防止在进行多线程操作时，遇到线程安全问题，确保数据的一致性和正确性
            // 先从文件中读取对应的 Message 数据
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);

                // 把读出来的 二进制数据转换成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);

                // 把 isVaild 设置成无效
                diskMessage.setIsValid((byte) 0x0);

                // 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            // 更新 数据记录文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0){
                stat.validCount -= 1;
            }
            writeStat(queue.getName(),stat);
        }

    }


    // 使用这个方法, 从文件中, 读取出所有的消息内容, 加载到内存中(具体来说是放到一个链表里)
    // 这个方法, 准备在程序启动的时候, 进行调用.
    // 这里使用一个 LinkedList, 主要目的是为了后续进行头删操作.
    // 这个方法的参数, 只是一个 queueName 而不是 MSGQueue 对象. 因为这个方法不需要加锁, 只使用 queueName 就够了.
    // 由于该方法是在程序启动时调用, 此时服务器还不能处理请求呢~~ 不涉及多线程操作文件.
    public LinkedList<Message> loadAllMessageFromQueue (String queueName) throws IOException, MessageQueueException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName)) ) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                // 这个变量记录当前文件光标.
                long currentOffset = 0;

                // 一个文件中包含了很多消息, 此处势必要循环读取.
                while (true) {

                    // 1. 读取当前消息的长度, 这里的 readInt 可能会读到文件的末尾(EOF)
                    //    readInt 方法, 读到文件末尾, 会抛出 EOFException 异常. 这一点和之前的很多流对象不太一样.
                    int messageSize = dataInputStream.readInt(); // 读取 信息数据之前的四个字节 里信息的长度

                    // 2. 按照这个长度, 读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize){
                        // 如果不匹配, 说明文件有问题, 格式错乱了!!
                        throw new MessageQueueException("[MessageFileManager] 文件格式错误! queueName = " + queueName);
                    }

                    // 3. 把这个读到的二进制数据, 反序列化回 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);

                    // 4. 判定一下看看这个消息对象, 是不是无效对象.
                    if (message.getIsValid() != 0x1) {
                        // 无效数据, 直接跳过.
                        // 虽然消息是无效数据, 但是 offset 不要忘记更新.
                        currentOffset += (4 + messageSize);
                        continue;
                    }

                    // 5. 有效数据, 则需要把这个 Message 对象加入到链表中. 加入之前还需要填写 offsetBeg 和 offsetEnd
                    //    进行计算 offset 的时候, 需要知道当前文件光标的位置的. 由于当下使用的 DataInputStream 并不方便直接获取到文件光标位置
                    //    因此就需要手动计算下文件光标.
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 +messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 这个 catch 并非真是处理 "异常", 而是处理 "正常" 的业务逻辑. 文件读到末尾, 会被 readInt 抛出该异常.
                // 这个 catch 语句中也不需要做啥特殊的事情
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }

        }
        return messages;
    }


    // 检查当前是否要针对该队列的消息数据文件进行 GC(垃圾回收)
    public boolean checkGC(String queueName){
        // 判定是否要 GC, 是根据总消息数和有效消息数. 总消息数大于 2000 并且 有效消息数 占 总消息数的 50%
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && ((double) stat.validCount / (double) stat.totalCount) > 0.5) {
            return true;
        }
        return false;
    }

    // 创建有一个新的 消息数据文件 用于进行对消息数据文件的垃圾回收工作
    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    // 这个方法,使用复制算法来完成. 真正执行消息数据文件的垃圾回收操作.
    public void gc(MSGQueue queue) throws MessageQueueException, IOException, ClassNotFoundException {
        // 进行 gc 的时候, 是针对消息数据文件进行大洗牌. 在这个过程中, 其他线程不能针对该队列的消息文件做任何修改.
        synchronized (queue){
            // 由于 gc 操作可能比较耗时, 此处统计一下执行消耗的时间. gc 执行开始的时间
            long gcBeg = System.currentTimeMillis();
            // 创建一个新的文件, 名字就是 queue_data_new.txt
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()){
                // 正常情况下, 这个文件不应该存在. 如果存在, 就是意外~~ 说明上次 gc 了一半, 程序意外崩溃了.
                throw new MessageQueueException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在! queueName = " + queue.getName());
            }
            boolean creatOK = queueDataNewFile.createNewFile();
            if (!creatOK){
                throw  new MessageQueueException("[MessageFileManager] 创建文件失败! queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }

            // 从旧的文件中, 读取出所有的有效消息对象了
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            // 把有效消息, 写入到新的文件中.
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        // 先写四个字节消息的长度
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }

            // 删除旧的数据文件, 并且把新的文件进行重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            boolean deleteOK = queueDataOldFile.delete();
            if (!deleteOK) {
                throw  new MessageQueueException("[MessageFileManager] gc删除数据旧文件失败! queueDataOldFile = " +queueDataOldFile.getAbsolutePath());
            }

            // 把 queue_data_new.txt => queue_data.txt
            boolean renameOK = queueDataNewFile.renameTo(queueDataOldFile);
            if (!renameOK){
                throw  new MessageQueueException("[MessageFileManager] gc重命名信数据文件失败! queueDataNewFile = " +queueDataNewFile.getAbsolutePath() + "queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }

            // 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            // gc 执行结束的时间
            long gcEnd = System.currentTimeMillis();
            // gc 耗时多久
            System.out.println("[MessageFileManager] gc 执行完毕! queueName = " + queue.getName() + ", time = "
                    + (gcEnd - gcBeg) + "ms");
        }
    }
}